# -*- coding: utf-8 -*-
"""
@Author : Chan ZiWen
@Date : 2022/6/9 10:12
File Description:

对表中所有数据进行更新，
"""
import os
import yaml
import time
import vthread
import json
import requests
import numpy as np
import pandas as pd

from datetime import datetime
from clickhouse_driver import Client

work_dir = os.path.dirname(__file__)

"""
id	mode	tv_mac	active_id	mac	current_time	analysis_datetime	final_15d_time	create_time	last_update_time
1	1	ad	86105349	sa	1653513595	1653513500	08:20				
"""
interval_15 = 1296000

# 用于并行处理数据
pool1 = vthread.pool(40, gqueue=1)  # 开40个伺服线程，组名为1
update_datas = []


def post(datas, url, headers):
    datas = json.dumps(datas)
    response = json.loads(requests.post(url, data=datas, headers=headers).text)
    if response['code'] != 1000:
        raise RuntimeError(f" {date}({response})")


def save2ck():
    """
    [
    {"mode":1,"activeId":"34","mac":"3243","tvMac":"23423","oneDayTime":15013452,"final15dTime":"12:00","analysisDatetime":"2022-02-12"},
    {"mode":1,"activeId":"34","mac":"3243","tvMac":"23423","oneDayTime":15013452,"final15dTime":"12:00","analysisDatetime":"2022-02-12"},
    ]
    :return:
    """
    url = args["url1_inner"]
    headers = {"Content-Type": "application/json"}
    inter_10000 = 10000
    n = len(update_datas)
    if n > 0:
        # 超过1w条：则分批存
        if n > inter_10000:
            nb = n // inter_10000
            for i in range(nb):
                datas_part = update_datas[i*inter_10000: (i+1)*inter_10000]
                post(datas_part, url, headers)
            if len(update_datas[(i+1)*inter_10000:]) > 0:
                post(update_datas[(i+1)*inter_10000:], url, headers)
        else:
            post(update_datas, url, headers)


def func(timestamp: int):
    """ 提取 时分 """
    date_time = datetime.fromtimestamp(timestamp)
    hour = date_time.hour
    min = date_time.minute
    if hour < 3:
        hour += 24
    return hour * 60 + min


@pool1
def utils(infos: list, res_days: list, analysis_date: str):
    # infos [mode, tv_mac, active_id, mac]
    """ 初始一个时间方便计算时间平均值 """
    if len(res_days) <= 0:
        print(ValueError(f"The value({res_days}) is Error!"))
        return
    elif len(res_days) == 1:
        final_time = res_days[0]
        datetime_c = datetime.fromtimestamp(final_time)
        final_time = f'{datetime_c.hour:>02d}:{datetime_c.minute:>02d}'
    elif len(res_days) == 2:
        # 提取 时分
        res_days = sorted(res_days)
        final_time = list(map(func, res_days))
        final_time = sum(final_time) // 2   # to integer
        hour, minute = (final_time // 60) % 24, final_time % 60
        final_time = f'{hour:>02d}:{minute:>02d}'
    else:
        # remove maximum and minimum
        res_days = sorted(res_days)
        if len(res_days) == 3:
            final_time = res_days[1]
            datetime_c = datetime.fromtimestamp(final_time)
            final_time = f'{datetime_c.hour:>02d}:{datetime_c.minute:>02d}'
        else:   # 取中位数
            # 提取 时分
            final_time = list(map(func, res_days))
            final_time = int(np.median(final_time[1:-1]))
            hour, minute = (final_time // 60) % 24, final_time % 60
            final_time = f'{hour:>02d}:{minute:>02d}'

    update_datas.append({
        'mode': int(infos[0]),
        'tvMac': infos[1],
        'activeId': infos[2],
        'mac': infos[3],
        'oneDayTime': int(res_days[-1]),
        'final15dTime': final_time,
        'analysisDatetime': str(analysis_date)})


def main(date_stamp, date_new):
    """ 根据 activeId 和 mac ，被扫描到的mac """
    sql = f"SELECT mode, tv_mac, active_id, mac, one_day_time, final_15d_time, toString(analysis_datetime), toUnixTimestamp(CONCAT(CAST(analysis_datetime as String), ' 00:00:00')) as ut " \
          f"FROM {args['config_ck']['table']} " \
          f"where ut >= {date_stamp - args['interval_15']} and ut <= {date_stamp}"
    data = client.execute(sql)
    if len(data) <= 0:
        return

    df_data = pd.DataFrame(data, columns=['mode', 'tv_mac', 'active_id', 'mac', 'one_day_time', 'final_15d_time', 'analysis_datetime', 'ut'])
    df_data = df_data.groupby(by=['mode', 'tv_mac', 'active_id', 'mac'])

    for id, dt in df_data:
        dt = dt.sort_values(by='ut')
        # 限制只能计算当前天的结果，并只存储这天的结果
        if dt['analysis_datetime'].values[-1] == date_new:
            utils(id, dt['one_day_time'].values, dt['analysis_datetime'].values[-1])
    vthread.pool.wait(gqueue=1)
    # write to ck
    print(len(update_datas))
    save2ck()
    update_datas.clear()


if __name__ == '__main__':
    # param
    args = yaml.safe_load(open(os.path.join(work_dir, "lifetime.yaml")))
    # define mysql connect
    client = Client(
        host=args['config_ck']['host'],
        database=args['config_ck']['database'],
        user=args['config_ck']['user'],
        password=args['config_ck']['password'])

    date = args['date']
    date_end = "2022-06-30"
    date_stamp = int(time.mktime(time.strptime(date, "%Y-%m-%d")))
    date_stamp_end = int(time.mktime(time.strptime(date_end, "%Y-%m-%d")))

    oneday = 86400
    length = (date_stamp_end - date_stamp) // oneday
    for _ in range(length):
        date_stamp += oneday
        date_new = datetime.fromtimestamp(date_stamp).strftime("%Y-%m-%d")
        print(f"Begin of the ({date_new})")
        main(date_stamp, date_new)
        print(f"End of the ({date_new})")

    # 去重
    sql = f"optimize table {args['config_ck']['table']} final"
    data = client.execute(sql)
    # 276,993
